package com.learn.kafka.stream.config;

import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.streams.KafkaStreams;
import org.apache.kafka.streams.StreamsConfig;
import org.apache.kafka.streams.Topology;

import java.util.Properties;

/**
 * kafka 配置
 * @author:         MI
 * @email:          448341911@qq.com
 * @createTime:     2025/8/9 16:06
 * @updateUser:     MI
 * @updateTime:     2025/8/9 16:06
 * @updateRemark:   修改内容
 * @version:        1.0
 */
public class KafkaStreamProperties {

    protected Properties props;

    // 构造方法：初始化通用配置
    public KafkaStreamProperties() {
        props = new Properties();
        initDefaultConfig();
    }

    // 初始化默认配置（通用配置）
    protected void initDefaultConfig() {
        // 应用ID（子类可覆盖）
        props.put(StreamsConfig.APPLICATION_ID_CONFIG, "default-stream-app");
        // Kafka集群地址（子类可覆盖）
        props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.110.17:9092");
        // 默认键值序列化器
        props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass());
        props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass());
        // 提交间隔（可选配置）
        props.put(StreamsConfig.COMMIT_INTERVAL_MS_CONFIG, 1000);
    }

    // 提供修改配置的方法（子类可调用）
    protected void setConfig(String key, String value) {
        props.put(key, value);
    }

    // 根据拓扑创建KafkaStreams实例
    public KafkaStreams createKafkaStreams(Topology topology) {
        return new KafkaStreams(topology, props);
    }
}
